package com.hub.realtime.resource.service.impl;

import java.io.File;
import java.util.Arrays;
import java.util.List;
import java.util.stream.Collectors;

import cn.hutool.core.date.DateUtil;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
import com.hub.realtime.common.core.domain.model.ClusterInfo;
import com.hub.realtime.common.core.domain.model.LoginUser;
import com.hub.realtime.common.exception.UtilException;
import com.hub.realtime.common.utils.StringUtils;
import com.hub.realtime.common.utils.hadoop.HdfsUtil;
import com.hub.realtime.framework.service.ServiceBase;
import com.hub.realtime.resource.domain.ResApplication;
import com.hub.realtime.resource.domain.ResCluster;
import com.hub.realtime.resource.domain.ResModuleJar;
import com.hub.realtime.resource.dto.input.JarManagementQueryParam;
import com.hub.realtime.resource.dto.input.UploadQueryParam;
import com.hub.realtime.resource.service.IResApplicationService;
import com.hub.realtime.resource.service.IResClusterService;
import com.hub.realtime.resource.service.IResModuleJarService;
import com.hub.realtime.system.service.ISysConfigService;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.io.FileUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import com.hub.realtime.resource.mapper.ResJarManagementMapper;
import com.hub.realtime.resource.domain.ResJarManagement;
import com.hub.realtime.resource.service.IResJarManagementService;
import org.springframework.web.multipart.MultipartFile;

import static com.hub.realtime.common.constant.Constants.*;
import static com.hub.realtime.common.constant.HdfsConstants.*;

/**
 * Jar包管理Service业务层处理
 *
 * @author wupeng
 * @date 2022-01-14
 */
@Service
@Slf4j
public class ResJarManagementServiceImpl extends ServiceBase<ResJarManagementMapper, ResJarManagement> implements IResJarManagementService {

    @Autowired
    IResModuleJarService moduleJarService;

    @Autowired
    IResApplicationService applicationService;

    @Autowired
    IResClusterService clusterService;

    @Autowired
    private ISysConfigService configService;


    @Override
    public Page<ResJarManagement> list(JarManagementQueryParam param) {
        LambdaQueryWrapper<ResJarManagement> wrapper = createCommonWrapper(ResJarManagement::getUserId, ResJarManagement::getCreateTime);
        if (StringUtils.isNotEmpty(param.getFileName())) {
            wrapper.like(ResJarManagement::getFileName, param.getFileName());
        }
        if (StringUtils.isNotEmpty(param.getJarSource())) {
            wrapper.eq(ResJarManagement::getJarSource, param.getJarSource());
        }
        if (StringUtils.isNotEmpty(param.getJarType())) {
            wrapper.eq(ResJarManagement::getJarType, param.getJarType());
        }
        Page<ResJarManagement> pageList = getPageList(param, wrapper);
        if (!pageList.getRecords().isEmpty()) {
            pageList.getRecords().forEach(d -> {
                ResApplication app = applicationService.getById(d.getApplicationId());
                d.setApplicationName(app.getName());
            });
        }
        return pageList;
    }


    @Override
    public int saveInfo(ResJarManagement jarManagement) {
        try {
            LoginUser loginUser = getLoginUser();
            int res = -1;
            if (jarManagement.getId() != null) {
                jarManagement.setUpdateBy(loginUser.getUsername());
                jarManagement.setUpdateTime(DateUtil.date());
                res = baseMapper.updateById(jarManagement);
            } else {
                ResApplication application = applicationService.getBaseMapper().selectById(jarManagement.getApplicationId());
                String hdfsDir = "";
                if (jarManagement.getJarType().equals(JAR_TYPE_TASK)) {
                    hdfsDir = UPLOAD_TASK_DIR.concat("/").concat(application.getUid());
                }
                if (jarManagement.getJarType().equals(JAR_TYPE_DEP)) {
                    hdfsDir = UPLOAD_DEP_DIR.concat("/").concat(application.getUid());
                }
                if (jarManagement.getJarType().equals(JAR_TYPE_UDF)) {
                    hdfsDir = UPLOAD_UDF_DIR.concat("/").concat(application.getUid());
                }
                ClusterInfo clusterInfo = applicationService.getClusterInfoByAppId(jarManagement.getApplicationId());
                if (!HdfsUtil.exists(clusterInfo, hdfsDir)) {
                    HdfsUtil.mkdirs(clusterInfo, hdfsDir);
                }
                String hdfsJarPath = HdfsUtil.getDefaultFS(clusterInfo).concat(hdfsDir.concat("/"));
                if (jarManagement.getJarSource().equals("git")) {
                    Long gitJarId = jarManagement.getGitJarId();
                    ResModuleJar resModuleJar = moduleJarService.getBaseMapper().selectById(gitJarId);
                    HdfsUtil.copyHdfs(clusterInfo, resModuleJar.getPath(), hdfsDir, false, true);
                    jarManagement.setFileName(resModuleJar.getName());
                    hdfsJarPath = hdfsJarPath.concat(resModuleJar.getName());
                } else {
                    String tmpDir = jarManagement.getTmpPath().substring(0, jarManagement.getTmpPath().length() - jarManagement.getFileName().length());
                    HdfsUtil.copyHdfs(clusterInfo, jarManagement.getTmpPath(), hdfsDir, false, true);
                    hdfsJarPath = hdfsJarPath.concat(jarManagement.getFileName());
                    HdfsUtil.delete(clusterInfo, tmpDir);
                }
                LambdaQueryWrapper<ResJarManagement> queryWrapper = new LambdaQueryWrapper<>();
                queryWrapper.eq(ResJarManagement::getApplicationId, jarManagement.getApplicationId());
                queryWrapper.eq(ResJarManagement::getFileName, jarManagement.getFileName());
                ResJarManagement currentJar = baseMapper.selectOne(queryWrapper);
                jarManagement.setFileName(jarManagement.getFileName());
                jarManagement.setFilePath(hdfsJarPath);
                jarManagement.setCreateBy(loginUser.getUsername());
                jarManagement.setCreateTime(DateUtil.date());
                if (currentJar == null) {
                    res = baseMapper.insert(jarManagement);
                } else {
                    Long id = currentJar.getId();
                    currentJar = jarManagement;
                    currentJar.setId(id);
                    res = updateById(currentJar) ? 1 : 0;
                }
            }
            return res;
        } catch (Exception ex) {
            log.error("创建jar包失败：" + ex.getMessage());
            ex.printStackTrace();
            throw new UtilException("创建jar包失败：" + ex.getMessage());
        }
    }

    @Override
    public int deleteInfos(Long[] ids) {
        List<ResJarManagement> jarManagements = baseMapper.selectBatchIds(Arrays.stream(ids).collect(Collectors.toList()));
        boolean hasInuse = false;
        if (jarManagements != null && !jarManagements.isEmpty()) {
            for (ResJarManagement jarManagement : jarManagements) {
                if (jarManagement.isInUse()) {
                    hasInuse = true;
                    break;
                }
            }
        }
        if (hasInuse) {
            throw new UtilException("存在正在使用的jar文件，不允许删除，请选择未使用的jar包进行删除！");
        }
        if (jarManagements != null && jarManagements.size() > 0) {
            jarManagements.forEach(jar -> {
                ResApplication application = applicationService.getBaseMapper().selectById(jar.getApplicationId());
                ResCluster cluster = clusterService.getBaseMapper().selectById(application.getClusterId());
                ClusterInfo clusterInfo = new ClusterInfo();
                clusterInfo.setHadoopConfDir(cluster.getHadoopDir());
                clusterInfo.setKerberosConfDir(cluster.getKererosDir());
                HdfsUtil.delete(clusterInfo, jar.getFilePath());
            });
        }
        return delByIds(ids);
    }

    /**
     * 创建上传的临时文件
     *
     * @param file
     * @param applicationId
     * @return
     */
    @Override
    public String createUploadFile(MultipartFile file, Long applicationId) {

        try {
            ClusterInfo clusterInfo = applicationService.getClusterInfoByAppId(applicationId);
            String tmpPath = configService.selectConfigByKey("sys:temp:dir").concat("/upload");
            File tmpFile = new File(tmpPath);
            if (!tmpFile.exists()) {
                tmpFile.mkdirs();
            }
            String fileName = file.getOriginalFilename();
            File localFile = new File(tmpPath.concat("/")
                    .concat(fileName));
            if (localFile.exists() && localFile.isFile()) {
                localFile.delete();
            }
            FileUtils.writeByteArrayToFile(localFile, file.getBytes());
            com.streamxhub.streamx.common.util.FileUtils.exists(localFile.getAbsolutePath());
            if (!HdfsUtil.exists(clusterInfo, UPLOAD_TMP_PATH)) {
                HdfsUtil.mkdirs(clusterInfo, UPLOAD_TMP_PATH);
            }
            HdfsUtil.upload(clusterInfo, localFile.getAbsolutePath(), UPLOAD_TMP_PATH, true, true);
            return HdfsUtil.getDefaultFS(clusterInfo).concat(UPLOAD_TMP_PATH.concat("/").concat(fileName));
        } catch (Exception ex) {
            log.error("上传文件出错：" + ex.getMessage());
            ex.printStackTrace();
            throw new UtilException("上传文件出错：" + ex.getMessage());
        }
    }

    @Override
    public void removeHdfsTmp(UploadQueryParam uploadQueryParam) {
        executorService.execute(() -> {
            try {
                ClusterInfo clusterInfo = applicationService.getClusterInfoByAppId(uploadQueryParam.getApplicationId());
                HdfsUtil.delete(clusterInfo, uploadQueryParam.getUploadPath());
            } catch (Exception ex) {

            }
        });
    }

    @Override
    public List<ResJarManagement> getJarListByAppId(Long applicationId) {
        LambdaQueryWrapper<ResJarManagement> queryWrapper = new LambdaQueryWrapper<>();
        queryWrapper.eq(ResJarManagement::getApplicationId, applicationId);
        queryWrapper.eq(ResJarManagement::getJarType, "task");
        return baseMapper.selectList(queryWrapper);
    }
}
